package cn.seecoder.ai.serviceImpl;

import cn.seecoder.ai.dao.FileInfoRepository;
import cn.seecoder.ai.dao.PredictionRepository;
import cn.seecoder.ai.enums.PredictionStateEnum;
import cn.seecoder.ai.exception.AIExternalException;
import cn.seecoder.ai.model.po.Prediction;
import cn.seecoder.ai.model.vo.*;
import cn.seecoder.ai.service.*;
import cn.seecoder.ai.utils.POJOConverter;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import scala.Tuple2;

import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @author fanyanpeng
 * @date 2023/4/21 5:11
 */
@Service
@Slf4j
public class PredictionServiceImpl implements PredictionService {
    @Autowired
    private FileInfoService fileInfoService;

    @Autowired
    PredictionRepository predictionRepository;

    @Autowired
    FeatureEngineeringService featureEngineeringService;

    @Autowired
    PredictionScalaService predictionScalaService;

    @Autowired
    ModelService modelService;

    @Autowired
    SparkSession sparkSession;

    @Autowired
    ApplicationContext applicationContext;

    private Prediction toPO(PredictionVO predictionVO){
        return (Prediction) POJOConverter.convert(predictionVO);
    }
    private PredictionVO toVO(Prediction prediction){
        return (PredictionVO) POJOConverter.convert(prediction);
    }



    /**
     * 获得属于某个用户的所有预测结果
     * @param userId
     * @return
     */
    @Override
    public List<PredictionVO> getPredictionVOList(Integer userId) {
        return predictionRepository.findByUserId(userId)
                .stream()
                .map(this::toVO)
                .collect(Collectors.toList());
    }

    /**
     * 某个用户下，使用某个模型的所有预测结果
     * @param userId
     * @param modelId
     * @return
     */
    @Override
    public List<PredictionVO> getPredictionVOList(Integer userId, Integer modelId) {
        return predictionRepository.findByUserIdAndModelId(userId,modelId)
                .stream()
                .map(this::toVO)
                .collect(Collectors.toList());

    }

    /**
     * 获取某一个预测结果
     * @param userId
     * @param predictionId
     * @return
     */
    @Override
    public PredictionVO getPredictionVO(Integer userId, Integer predictionId) {
        Optional<Prediction> predictionOptional = predictionRepository.findById(predictionId);
        if(!predictionOptional.isPresent()){
            throw AIExternalException.predictionNotFound();
        }
        Prediction prediction=predictionOptional.get();
        return toVO(prediction);
    }

    /**
     * 异步返回
     * 先创建，后补充值
     * @param userId
     * @param modelId
     * @param fileInfoId
     * @return
     */
    @Override
    public PredictionVO predictAsync(Integer userId, Integer modelId, Integer fileInfoId) {

        ModelVO modelVO = modelService.getModelVO(userId, modelId);
        String modelUri = modelVO.getModelUri();
        PipelineModel pipelineModel = PipelineModel.load(modelUri);

        PredictionVO predictionVO = createPrediction(userId, modelId,fileInfoId);

        // 异步调用
        PredictionService predictionService = applicationContext.getBean(PredictionService.class);
        predictionService.predictWithPipeLineModelAsync(userId,modelId,pipelineModel,fileInfoId,predictionVO.getId());

        return predictionVO;
    }


    /**
     * 最终调用的模型预测过程，中间设计模型预测的各种状态
     * @author   fanyanpeng
     * @date 2023/4/22 22:14
     * @param userId
     * @param modelId
     * @param pipelineModel
     * @param fileInfoId
     * @param predictionId
     * @return void
     */
    @Async
    public void predictWithPipeLineModelAsync(Integer userId,
                                       Integer modelId,
                                       PipelineModel pipelineModel,
                                       Integer fileInfoId,
                                       Integer predictionId){
        log.info("异步使用模型进行预测");
        predictWithPipeLineModelSync(userId,modelId,pipelineModel,fileInfoId,predictionId);
    }


    /**
     * 最终调用的模型预测过程，中间设计模型预测的各种状态
     * 以同步的方式进行预测，返回预测结果
     * @param userId
     * @param modelId
     * @param pipelineModel
     * @param fileInfoId
     * @param predictionId
     */
    @Override
    public PredictionVO predictWithPipeLineModelSync(Integer userId,
                                                     Integer modelId,
                                                     PipelineModel pipelineModel,
                                                     Integer fileInfoId,
                                                     Integer predictionId){
        // 如果没有预测结果，先创建一个预测结果
        if(predictionId == null){
            predictionId = createPrediction(userId, modelId, fileInfoId).getId();
        }

        Prediction prediction = getPrediction(predictionId);
        try{
            // 开始预测
            prediction.setPredictionState(PredictionStateEnum.ON_GOING);
            predictionRepository.save(prediction);

            // 调用预测函数，返回预测结果uri和示例数据
            Tuple2<String,String> predictResult  = predictionScalaService.usePipeLineModelToPredict(fileInfoId,modelId,pipelineModel);
            String predictionUri = predictResult._1;
            String exampleData = predictResult._2;

            log.info("predictionUri = {}",predictionUri);
            log.info("exampleDate = {}",exampleData);

            // 修改状态为成功
            prediction.setPredictionState(PredictionStateEnum.SUCCESS);
            prediction.setFinishTime(new Date());
            prediction.setExampleData(exampleData);
            prediction.setPredictionFileDownloadingUri(predictionUri);

            predictionRepository.save(prediction);

        }catch (Exception e){
            e.printStackTrace();
            // 设置预测失败
            prediction.setPredictionState(PredictionStateEnum.FAIL);
            predictionRepository.save(prediction);
        }

        return getPredictionVO(userId, predictionId);

    }

    /**
     * 生成预测结果，但是不进行预测，只是创建一个预测结果，返回预测结果的id
     *
     * @param userId
     * @param modelId
     * @param fileInfoId
     * @return
     */
    public PredictionVO createPrediction(Integer userId, Integer modelId, Integer fileInfoId) {
        ensurePredictFileStructureMatchesTrainingFileStructure(modelId, fileInfoId);
        Prediction prediction = Prediction.builder()
                .predictionState(PredictionStateEnum.WAITING)
                .createdTime(new Date())
                .modelId(modelId)
                .userId(userId)
                .fileInfoId(fileInfoId)
                .build();

        predictionRepository.save(prediction);

        return getPredictionVO(userId, prediction.getId());

    }

    @Autowired
    ConfigService configService;

    private void ensurePredictFileStructureMatchesTrainingFileStructure(Integer modelId, Integer fileInfoId) {
        ModelVO modelVO = modelService.getModelVO(0,modelId);
        ConfigVO configVO = configService.getConfig(0,modelVO.getConfigId());
        List<String> features = configVO.getFeatures();
        FileInfoVO trainFileInfoVO = fileInfoService.getFileInfo(0,configVO.getFileInfoId());
        FileInfoVO predictFileInfoVO = fileInfoService.getFileInfo(0,fileInfoId);

        List<FileFieldVO> trainFieldVOList = trainFileInfoVO.getFileFields().stream()
                .filter(fileFieldVO -> features.contains(fileFieldVO.getFieldName()))
                .collect(Collectors.toList());
        List<FileFieldVO> predictFieldVOList = predictFileInfoVO.getFileFields().stream()
                .filter(fileFieldVO -> features.contains(fileFieldVO.getFieldName()))
                .collect(Collectors.toList());
        if(!sameStructure(trainFieldVOList,predictFieldVOList)){
            throw AIExternalException.predictFileStructureNotMatchTrainFileStructure();
        }


    }

    /**
     * 判断两个文件结构是否相同
     * @param fieldVOList1
     * @param fieldVOList2
     * @return
     */
    private Boolean sameStructure(List<FileFieldVO> fieldVOList1, List<FileFieldVO> fieldVOList2){
        if(fieldVOList1.size() != fieldVOList2.size()){
            return false;
        }

        fieldVOList1.sort(Comparator.comparing(FileFieldVO::getFieldName));
        fieldVOList2.sort(Comparator.comparing(FileFieldVO::getFieldName));

        for(int i=0;i<fieldVOList1.size();i++){
            FileFieldVO fileFieldVO1 = fieldVOList1.get(i);
            FileFieldVO fileFieldVO2 = fieldVOList2.get(i);
            if(!fileFieldVO1.getFieldName().equals(fileFieldVO2.getFieldName())){
                return false;
            }
            if(!fileFieldVO1.getDataType().equals(fileFieldVO2.getDataType())){
                return false;
            }
        }
        return true;
    }


    private Prediction getPrediction(Integer predictionId){
        return predictionRepository.findById(predictionId).get();
    }
}
